package com.coolingme.consumer.cyclicsync;

import cn.hutool.core.thread.ThreadUtil;
import com.coolingme.KafkaAppTest;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.Test;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * kafka循环消费和CyclicBarrier结合，使用多线程处理消费的数据
 *
 * @author wangyue
 * @date 2023/12/10 22:51
 */
@Slf4j
public class ConsumerSyncDemoTest2 extends KafkaAppTest {

    private KafkaConsumer<?, ?> consumer;

    // 已经启动的线程数量
    private AtomicInteger currentThreadNum = new AtomicInteger(0);

    // CyclicBarrier允许的最大运行线程数量
    private int maxThreadNum = 3;

    // 创建线程池
    private ExecutorService executorService = Executors.newFixedThreadPool(maxThreadNum * 2);

    private CyclicBarrier cyclicBarrier;

    private CyclicBarrierRunnable cyclicBarrierRunnable;

    private KafkaConsumer<?, ?> createConsumer(String bootstrapServers, String groupId) {
        // 配置信息
        Properties properties = new Properties();
        // 连接的集群
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        // key,value反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        /*properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");*/

        //关闭自动提交OFFSET，默认值为true
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        //一次拉取一条数据，默认500条
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
        // 消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        // 重置消费者的offset
        // 当前消费者组不存在或服务上没有数据（如：数据被删除）
        // earliest:重置为最早的offset，获取最早的数据
        // latest:重置为最新的offset，获取最新的数据
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 设置日志级别为 INFO
//        properties.put("log4j.rootLogger", "INFO, stdout");

        // 消费者
        return new KafkaConsumer<>(properties);
    }

    private void loadConsumer() {
        String bootstrapServers = "192.168.112.128:9092";

        String topic = "test-topic";

        String groupId = "test-topic-group";

        Integer partition = 0;

        this.consumer = createConsumer(bootstrapServers, groupId);

        // 订阅主题,若主题不存在会自动创建
        TopicPartition topicPartition = new TopicPartition(topic, partition);
        this.consumer.assign(Arrays.asList(topicPartition));
        log.info("消费者加载成功：{}", this.consumer);
    }

    @Test
    public void test() {

        loadConsumer();

        loadCyclicBarrierTask();

        pollData();

        // 让测试类先别结束
        ThreadUtil.sleep(1, TimeUnit.DAYS);
    }


    private void loadCyclicBarrierTask() {
        // 由最后一个线程执行CyclicBarrier的barrierAction
        this.cyclicBarrier = new CyclicBarrier(this.maxThreadNum, () -> {
            // 所有线程都完成后，将计数器重置为0
            this.currentThreadNum.set(0);
            log.info("{},========所有线程已完成，开始重新执行任务========\n", Thread.currentThread().getName());

            // 同步提交，当前线程会阻塞，直到提交offset成功
            this.consumer.commitSync();

            ThreadUtil.sleep(3, TimeUnit.SECONDS);

            // 重新从kafka中拉取新数据
            pollData();

        });

//        this.cyclicBarrierRunnable = new CyclicBarrierRunnable(cyclicBarrier);
    }


    private void pollData() {
        while (this.currentThreadNum.get() < this.maxThreadNum) {
//            log.info("-------------休眠-------------");
//            ThreadUtil.sleep(3 * 1000L);
            log.info("---------开始拉取数据----------");
            // 拉取数据
            ConsumerRecords<?, ?> records = consumer.poll(Duration.ofMillis(3 * 1000));
            if (records.isEmpty()) {
                log.info("当前无数据...");
                continue;
            } /*else {
                log.info("共拉取 {} 条数据", records.count());
            }*/

            // 一个线程处理的一批数据
            List<ConsumerRecord<?, ?>> recordList = new ArrayList<>();

            // 根据配置，只会拉取一条数据
            for (ConsumerRecord<?, ?> record : records) {
//                log.info(StrUtil.format("消费到的数据：topic= {},partition= {},offset= {},key= {},value= {}",
//                        record.topic(), record.partition(), record.offset(), record.key(), record.value()));
                recordList.add(record);
            }

            this.executeTask(recordList);

        }
    }


    private void executeTask(List<ConsumerRecord<?, ?>> recordList) {

//        this.cyclicBarrierRunnable.setRecordList(recordList);

        // 已运行的线程数量+1
        this.currentThreadNum.incrementAndGet();
        log.info("当前线程数：{}", this.currentThreadNum.get());

        // 线程池执行任务
        this.executorService.submit(new CyclicBarrierRunnable(this.cyclicBarrier, recordList));

        /*this.executorService.submit(() -> {
            TimeInterval timerAll = DateUtil.timer();
            TimeInterval timer = DateUtil.timer();
            log.info("{} -正在准备...", Thread.currentThread().getName());
            try {
                for (ConsumerRecord<?, ?> record : recordList) {
                    log.info(StrUtil.format("处理数据1：topic= {},partition= {},offset= {},key= {},value= {}",
                            record.topic(), record.partition(), record.offset(), record.key(), record.value()));
                }

                // 随机等待,模拟处理时间
                ThreadUtil.sleep(RandomUtil.randomInt(3, 10), TimeUnit.SECONDS);

                for (ConsumerRecord<?, ?> record : recordList) {
                    log.info(StrUtil.format("处理数据2：topic= {},partition= {},offset= {},key= {},value= {}",
                            record.topic(), record.partition(), record.offset(), record.key(), record.value()));
                }

                log.info("{} -处理完成，等待所有线程完成，耗时：{}", Thread.currentThread().getName(), timer.intervalPretty());
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
            ThreadUtil.sleep(RandomUtil.randomInt(3, 10), TimeUnit.SECONDS);
            log.info("{} -结束，耗时：{}", Thread.currentThread().getName(), timerAll.intervalPretty());

        });*/



        /*class ARunnable implements Runnable {
            @Override
            public void run() {
                System.out.println(1);
            }
        }

        this.executorService.submit(new ARunnable());

        this.executorService.submit(() -> {
            System.out.println(1);
        });*/
    }


}

